package beautiful.butterfly.drds.binlog.parser;


import beautiful.butterfly.drds.binlog.connection.ConnectionAttributes;
import beautiful.butterfly.drds.binlog.event.EventData;
import beautiful.butterfly.drds.binlog.event.data.TableMapEventData;
import beautiful.butterfly.drds.binlog.event.data.UpdateRowsEventData;
import beautiful.butterfly.drds.binlog.netty.server.NioSocketChannelExtend;
import beautiful.butterfly.drds.binlog.utils.ByteUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.Serializable;
import java.util.*;

@Slf4j
public class UpdateRowsEventDataParser implements EventDataParser
{


    private boolean mayContainExtraInformation = false;

    public UpdateRowsEventDataParser(boolean b)
    {
        mayContainExtraInformation = b;
    }

    @Override
    public EventData parse(ByteBuf msg, ChannelHandlerContext context, int checksumLength)
    {
        // LoggerUtils.debug(logger, "");
        // LoggerUtils.debug(logger,
        // "enter UpdateRowsEventDataParser,mayContainExtraInformation:" +
        // mayContainExtraInformation);
        // LoggerUtils.debug(logger, "enter
        // UpdateRowsEventDataParser.parse...");
        long tableId = ByteUtils.readUnsignedLong(msg, 6);
        msg.skipBytes(2);
        if (mayContainExtraInformation)
        {
            int extraInfoLength = ByteUtils.readUnsignedInt(msg, 2);
            msg.skipBytes(extraInfoLength - 2);
        }
        // 继续处理
        int numberOfColumns = ByteUtils.readVariableNumber(msg).intValue();
        BitSet includedColumnsBeforeUdpate = null;
        try
        {
            includedColumnsBeforeUdpate = ByteUtils.readBitSet(msg, numberOfColumns, true);
        } catch (IOException e)
        {
            e.printStackTrace();
        }
        BitSet includedColumns = null;
        try
        {
            includedColumns = ByteUtils.readBitSet(msg, numberOfColumns, true);
        } catch (IOException e)
        {
            e.printStackTrace();
        }
        // rows
        List<Map.Entry<Serializable[], Serializable[]>> rows = new ArrayList<Map.Entry<Serializable[], Serializable[]>>();
        ConnectionAttributes connectionAttributes = ((NioSocketChannelExtend) context.channel()).getConnectionAttributes();
        TableMapEventData tableMapEventData = connectionAttributes.getTableMapEventData(tableId);
        if (null == tableMapEventData)
        {
            log.info("error,fail to find tableMapEventData ,tableId is:" + tableId);
        }
        String database = tableMapEventData.getDatabaseName();
        String table = tableMapEventData.getTableName();
        try
        {
            while (msg.readableBytes() > checksumLength)
            {// 表明还有内容可取
                // LoggerUtils.debug(logger, "还有数据...");
                rows.add(new AbstractMap.SimpleEntry<Serializable[], Serializable[]>(
                        ByteUtils.deserializeRow(tableMapEventData, includedColumnsBeforeUdpate, msg),
                        ByteUtils.deserializeRow(tableMapEventData, includedColumns, msg)));
            }

        } catch (IOException e)
        {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        // 构造
        UpdateRowsEventData eventData = new UpdateRowsEventData();
        eventData.setTableId(tableId);
        eventData.setTableMapEventData(tableMapEventData);// 为了获取database&tableName
        if (connectionAttributes.acceptByFilter(database, table))
        {// 不处理的就不需要了
            HashMap<String, String> mappings = connectionAttributes.getPositionToColumnNameMap(database, table);
            eventData.setIncludedColumnsBeforeUpdate(includedColumnsBeforeUdpate, mappings);
            eventData.setIncludedColumns(includedColumns, mappings);
        }
        eventData.setRows(rows);
        return eventData;
    }

}
